// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: leon

#include "coin2/exchange/ftx_common/feed_v1/processor.h"

#include <string>
#include <string_view>
#include <utility>

#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/ftx_common/feed_v1/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::ftx_common::feed_v1::impl {

void FtxCommonFeedProcessor::OnTopicRecord(const base::executor::TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  rapidjson::Document document;
  document.Parse<rapidjson::kParseNumbersAsStringsFlag>(packet.payload.data());
  if (document.HasParseError()) {
    throw std::runtime_error(
        std::string("JSON parse error: ") + GetParseError_En(document.GetParseError()));
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  // type and channel from websocket API
  // success and resutl from REST API
  bool is_channel = false;
  bool is_rest_result = false;
  std::string_view native_symbol;
  FtxCommonChannel parsed_channel{"ignored"};
  if (document.HasMember("type") && document.HasMember("channel")) {
    is_channel = true;
    const std::string& channel = document["channel"].GetString();
    const std::string& type = document["type"].GetString();
    parsed_channel = FtxCommonChannel::Parse(type, channel);

    if (!CheckChannelFilter(parsed_channel.channel)) {
      upd_mgr_.onEndPacket(packet);
      return;
    }
    native_symbol = document["market"].GetString();
    (void) is_channel;
  } else if (document.HasMember("success") &&
             document["success"].GetBool() &&
             document.HasMember("result")) {
    is_rest_result = true;
  } else if (document.HasMember("url")) {
    std::string_view ch_candle = "/candles?resolution=";
    const std::string& url = document["url"].GetString();
    size_t candle_pos = url.find(ch_candle);
    std::string channel;
    if (candle_pos != std::string::npos) {
      parsed_channel.channel = "kline";
      is_channel = true;
      size_t sep_pos = url.find("&");
      if (sep_pos == std::string::npos) {
        channel = url.substr(candle_pos, sep_pos);
      } else {
        channel = url.substr(candle_pos, sep_pos - candle_pos);
      }
    }

    if (!CheckChannelFilter(channel)) {
      upd_mgr_.onEndPacket(packet);
      return;
    }
  }else {
    LOG(INFO) << "[Ftx Common Feed Processor] meta msg for " << packet.topic_name << ": "
              << packet.payload.data();
    upd_mgr_.onEndPacket(packet);
    return;
  }

  if (parsed_channel.channel == "orderbook") {
    auto* book = upd_mgr_.GetBookMutable(native_symbol);
    if (book == nullptr) {
      upd_mgr_.onEndPacket(packet);
      return;
    }
    const std::string& type = document["type"].GetString();
    if (type == "partial") {
      FtxCommonFeedUpdateParser::ParseSnapshot(book, document["data"]);
      upd_mgr_.PublishBook();
    } else if (type == "update") {
      FtxCommonFeedUpdateParser::ParseUpdate(book, document["data"]);
      upd_mgr_.PublishBook();
    } else {
      LOG(INFO) << "[Ftx Common Feed Processor] unknown msg for " << packet.topic_name << ": "
                << packet.payload.data();
    }
  } else if (parsed_channel.channel == "trades") {
    if (!upd_mgr_.is_symbol_registered(native_symbol)) {
      upd_mgr_.onEndPacket(packet);
      return;
    }

    if (parser_config_.has_trade_sampling_frequency_bps()) {
      auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
      for (const auto& json_trade : document["data"].GetArray()) {
        Trade _trade{};
        FtxCommonFeedUpdateParser::ParseTrade(json_trade, &_trade);
        bool aggregated = slim_trade->TryAggregateSlimTrade(
                packet.timestamp,
                _trade.price,
                _trade.fill_qty,
                _trade.side,
                _trade.timestamp,
                parser_config_.trade_sampling_frequency_bps());
        int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
        if (!aggregated ||
            timespan > parser_config_.trade_sampling_time_length()) {
          auto* trade = upd_mgr_.GetTradeMutable();
          CHECK(slim_trade->PopulateTradeAndClear(trade));
          upd_mgr_.PublishTrade(native_symbol, false);
        }
      }
    } else {
      Trade* agg_trade = upd_mgr_.GetTradeMutable();
      agg_trade->Clear();
      Trade trade{};

      for (auto& trade_json : document["data"].GetArray()) {
        FtxCommonFeedUpdateParser::ParseTrade(trade_json, &trade);
        if (TryAggregateTrades(agg_trade, trade)) {
          continue;
        }
        if (agg_trade->fill_qty > 0) {
          upd_mgr_.PublishTrade(native_symbol, true);
        }
        *agg_trade = trade;
      }
      if (agg_trade->fill_qty > 0) {
        upd_mgr_.PublishTrade(native_symbol, false);
      }
    }
  } else if (parsed_channel.channel == "ticker") {
    auto* book = upd_mgr_.GetBookMutable(native_symbol);
    if (book == nullptr) {
      upd_mgr_.onEndPacket(packet);
      return;
    }

    FtxCommonFeedUpdateParser::ParseBbo(book, document["data"]);
    if (parser_config_.has_sampling_frequency_bps()) {
      bool sampled = book->SampleByFrequency(
                book->Bid0(),
                book->Ask0(),
                parser_config_.sampling_frequency_bps());
      if (parser_config_.has_trade_sampling_frequency_bps()) {
        auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
        int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
        if (sampled ||
            timespan > parser_config_.trade_sampling_time_length()) {
          auto* trade = upd_mgr_.GetTradeMutable();
          if (slim_trade->PopulateTradeAndClear(trade)) {
            upd_mgr_.PublishTrade(native_symbol, false);
          }
        }
      }
      if (sampled) {
        upd_mgr_.PublishBook();
      }
    } else {
      upd_mgr_.PublishBook();
    }

    // not implemented
    upd_mgr_.onEndPacket(packet);
    return;
  } else if (parsed_channel.channel == "kline") {
    // /markets/{market_name}/candles?resolution={resolution}
    std::string_view markets = "/markets/";
    std::string_view candles = "/candles?";
    const std::string& url = document["url"].GetString();
    auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
    size_t markets_pos = url.find(markets) + markets.size();
    size_t candles_pos = url.find(candles);
    std::string native_symbol = url.substr(markets_pos, candles_pos - markets_pos);
    auto interval = parameters["resolution"];
    fastfeed::proto::KlineInterval kinterval =
        FtxCommonFeedUpdateParser::ParseKlineInterval(interval);

    auto* kline = upd_mgr_.GetKlineMutable(native_symbol, kinterval);
    if (kline != nullptr) {
      int page = -1;
      if (parameters.count("page") > 0) {
        page = std::stoi(parameters["page"]);
      }
      const auto& kline_array = document["data"]["result"].GetArray();
      if (page == 0) {
        kline->SnapshotClear();
      } else {
        kline->UpdateClear();
      }
      bool to_publish = false;
      for (auto& kline_json : kline_array) {
        bool freezed = FtxCommonFeedUpdateParser::ParseKline(kline_json, kline, packet.timestamp,
            kinterval);
        to_publish = to_publish || freezed;
      }

      if (to_publish) upd_mgr_.PublishKline(native_symbol);
    }
  } else if (is_rest_result) {
    // 1 REST API result may contains several non_tbs channels data
    if (document["result"].IsArray()) {
      for (auto& result_json : document["result"].GetArray()) {
        // mark_price, open_interest and index is in the same raw feed message
        if (result_json.HasMember("mark") &&
            result_json.HasMember("index") &&
            result_json.HasMember("openInterest")) {
          // share the same channel
          if (!CheckChannelFilter("index")) {
            upd_mgr_.onEndPacket(packet);
            return;
          }

          native_symbol = result_json["name"].GetString();
          if (!upd_mgr_.is_symbol_registered(native_symbol)) continue;

          auto* mark_price = upd_mgr_.GetMarkPriceMutable();
          mark_price->Clear();
          mark_price->timestamp = packet.timestamp;
          mark_price->mark_price = stod(result_json["mark"].GetString());
          upd_mgr_.PublishMarkPrice(native_symbol);

          auto* index = upd_mgr_.GetIndexMutable();
          index->Clear();
          index->timestamp = packet.timestamp;
          index->price = stod(result_json["index"].GetString());
          upd_mgr_.PublishIndex(native_symbol);

          auto* open_interest = upd_mgr_.GetOpenInterestMutable();
          open_interest->timestamp = packet.timestamp;
          open_interest->open_interest_qty = stod(result_json["openInterest"].GetString());
          upd_mgr_.PublishOpenInterest(native_symbol);
        } else if (result_json.HasMember("rate") && result_json.HasMember("future")) {
          if (CheckChannelFilter("funding_rate")) {
            native_symbol = result_json["future"].GetString();
            if (!upd_mgr_.is_symbol_registered(native_symbol)) continue;

            auto* funding_rate = upd_mgr_.GetFundingRateMutable();
            funding_rate->Clear();
            funding_rate->timestamp = packet.timestamp;
            funding_rate->funding_rate = stod(result_json["rate"].GetString());
            funding_rate->funding_time =
                TimestampFromFtxIso8601(result_json["time"].GetString());
            upd_mgr_.PublishFundingRate(native_symbol);
          }
        }
      }
    }

  } else {
    LOG(INFO) << "[Ftx Common Feed Processor] unknown msg for " << packet.topic_name << ": "
              << packet.payload.data();
  }

  upd_mgr_.onEndPacket(packet);
}  // namespace coin2::exchange::ftx_common::feed_v1::impl
}  // namespace coin2::exchange::ftx_common::feed_v1::impl
